package edu.qdu.offline

import com.mongodb.casbah.Imports.{MongoClientURI, MongoDBObject}
import com.mongodb.casbah.MongoClient
import org.apache.spark.mllib.recommendation.ALS
import org.apache.spark.mllib.recommendation.Rating
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.jblas.DoubleMatrix


/**
 * 离线推荐服务和相似商品的加工
 */

//封装评分数据
//封装用户商品评分数据
case class ProductRating(userId:Int,productId:Int,rating:Double,timestamp:Long)
//封装mongoDB的连接信息
case class MongoConfig(url:String,db:String)


//封装商品类
case class Recommendation(productId: Int,score:Double)
//封装产品相似矩阵
case class ProductRecs(productId:Int,recs:Seq[Recommendation])
//封装用户推荐列表
case class UserRecs(userId:Int,recs:Seq[Recommendation])

object OffLineRecommender {
  def main(args: Array[String]): Unit = {

    //Mongo的连接配置
    val configMap = Map(
      "url" -> "mongodb://spark01:27017/recommender",
      "db" -> "recommender"
    )

    //定义一个隐士参数
    implicit val config = MongoConfig(configMap.getOrElse("url", ""), configMap.getOrElse("db", ""))

    //定义集合常量
    val USER_RECS_COLLECTION = "UserRecs"   //用户推荐存储在MongoDB中的集合
    val PRODUCT_RECS_COLLECTION = "ProductRecs"   //商品相似矩阵存储在MongoDB中的集合
    val RATING_COLLECTION = "Rating"  //用户商品评分集合
    val MAX_RECOMMEND_NUMBER = 20   //最大推荐数量

    //获取SparkSQL的上下文
    val spark = SparkSession.builder()
      .appName("offline-recommender")
      .master("local[*]") //如果往服务器部署，需要修改一下即可
      .getOrCreate()

    //加载Ratings的数据，然后转换成RDD，并将其进行缓存
    import spark.implicits._
    val ratingRDD = spark
      .read
      .option("uri", config.url)
      .option("collection", RATING_COLLECTION)
      .format("com.mongodb.spark.sql")
      .load() //返回DataFrame
      .as[ProductRating] //将df转换成DS
      .rdd //将ds转换成RDD
      .map(line => (line.userId, line.productId, line.rating)) //构建用户、商品、打分三元组
      .cache()   //缓存
    //ratingRDD.foreach(println(_))
    //构建训练集 --- 封装的Rating一定是recommendation包下的
    val trainData = ratingRDD.map(x => org.apache.spark.mllib.recommendation.Rating(x._1, x._2, x._3))

    //获取userId和productId的去重列表
    val userRDD = ratingRDD.map(x => x._1).distinct()   //去重后的用户列表
    val productRDD = ratingRDD.map(x => x._2).distinct()  //去重后的商品列表
    //使用去重后的userRDD和productRDD进行笛卡尔积计算，形成一个空的用户商品矩阵
    val userProduct = userRDD.cartesian(productRDD)
    //userProduct.foreach(println(_))

    //构建ALS模型
    val (rank,iterations,regex) = (200,10,0.01)   //分别为隐士因子个数；迭代次数；正则化系数
    //引入mllib包下的ALS --->使用训练集来训练ALS模型
    val alsMode = ALS.train(trainData,rank,iterations,regex)

    //预测 --->使用userProduct来进行预测
    val preRDD = alsMode.predict(userProduct)

    //对预测结果集进行过滤和封装
    val userRecsDF = preRDD //RDD[Rating]
      .filter(x => x.rating > 0) //将预测的评分大于0的商品过滤出来
      .map(rating => {
        (rating.user, (rating.product, rating.rating)) //将数据封装成 (userid,(productId,rating))
      })
      .groupByKey() //按照user来进行分组 //(12，Iterable[(),(),(),(),()])
      .map(x => {
        //封装成最终的结果数据
        UserRecs(
          x._1, //用户Id
          x._2.toList //将Iterable[(),(),(),(),()]转换成List
            .sortWith(_._2 > _._2) //使用得分进行降序排序；；大于是降序
            .take(MAX_RECOMMEND_NUMBER) //获取得分最高的前20个商品
            .map(x => Recommendation(x._1, x._2)) //将商品Id和预测得分封装到Recommendation中
        )
      })
      .toDF()
    //userRecsDF.show()
    //将数据写出MongoDB中UserRecs集合中
    writeDataToMongoDB(userRecsDF,USER_RECS_COLLECTION)

    //商品相似矩阵
    //1、获取商品的特征向量 --->10维度
    val productFeatures:RDD[(Int,DoubleMatrix)] = alsMode.productFeatures
      .map(x => {
        (x._1, new DoubleMatrix(x._2))  //(商品ID，对应商品的特征向量)
      })
    //2、获取每一个商品的余弦相似度
    val productRecsDF = productFeatures.cartesian(productFeatures) //商品特征向量笛卡尔积
      .filter(x => x._1._1 != x._2._1) //过滤掉两个相同的商品
      .map(x => {
        val consin = consinSim(x._1._2, x._2._2) //使用商品的特征向量调用consinSim方法
        //封装返回
        (x._1._1, (x._2._1, consin)) //(p1商品Id，(p2商品Id,p1和p2的余弦相似度))
      })
      .filter(x => x._2._2 > 0.6) //过滤余弦相似度小于0.6的商品
      .groupByKey() //根据商品ID(x._1._1)来进行分组
      .map(x => {
        //封装返回的数据ProductRecs
        ProductRecs(
          x._1, //商品ID
          x._2
            .toList
            .map(x => Recommendation(x._1, x._2)) //封装(商品Id，余弦相似度)
        )
      })
      .toDF()
    //productRecsDF.show()

    //将商品的相似矩阵存储到MongoDB中，--->已被实时推荐需要
    writeDataToMongoDB(productRecsDF,PRODUCT_RECS_COLLECTION)

    //关闭spark
    spark.stop()
  }

  /**
   * 将productDF和ratingDF写入数库的集合中
   * @param spark
   * @param productDF
   * @param ratingDF
   * @param config
   */
  def writeDataToMongoDB(df: DataFrame,col:String)(implicit config:MongoConfig): Unit ={
    //获取Mongo的client
    val client = MongoClient(MongoClientURI(config.url))
    //获取操作集合
    val collection = client(config.db)(col)

    //删除集合
    collection.dropCollection()

    //将productDF写入collection中
    df
      .write
      .option("uri",config.url)
      .option("collection",col)
      .mode(SaveMode.Overwrite)
      .format("com.mongodb.spark.sql")
      .save()
  }

  /**
   * 余弦相似度
   * @param product1
   * @param product2
   */
  def consinSim(product1:DoubleMatrix,product2:DoubleMatrix): Double = {
    //相似公式：y = (DXi * DYi)的加和值 / (DXi平方和 + DYi平方和)   i属于(1-n)
    product1.dot(product2) / (product1.norm2() * product2.norm2())
  }
}